package Demo3;

import com.google.common.base.Optional;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.Time;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

import java.util.ArrayList;



/**
 * Created by lenovo on 2017/12/8.
 * 利用Streaming的Transform算子实现广告点击的黑名单过滤。
 */
public class Streaming_TransformJava {
    public static void main(String[] args){
       SparkConf conf = new SparkConf().setAppName("Streaming_TransformJava").setMaster("local[2]").set("spark.testing.memory","2147480000");
       JavaStreamingContext ssc = new JavaStreamingContext(conf, Durations.seconds(5));

       ArrayList<Tuple2<String,Boolean>> blockList = new ArrayList<Tuple2<String,Boolean>>();
        blockList.add(new Tuple2<String, Boolean>("zhangsan",true));
        blockList.add(new Tuple2<String, Boolean>("lisi",true));
       final JavaPairRDD<String,Boolean> blockRDD = ssc.sparkContext().parallelizePairs(blockList);

       JavaReceiverInputDStream userDStream = ssc.socketTextStream("hadoop1",2222);
       JavaPairDStream<String,String> userCloikDStream = userDStream.mapToPair(new PairFunction<String,String,String>() {
            @Override
            public Tuple2<String,String> call(String lines) throws Exception {
                return new Tuple2<String, String>(lines.split(" ")[1],lines);
            }
        });

      JavaDStream<String> userLogDStream = userCloikDStream.transform(new Function<JavaPairRDD<String, String>, JavaRDD<String>>() {
            @Override
            public JavaRDD<String> call(JavaPairRDD<String, String> userRDD) throws Exception {
              JavaPairRDD<String,Tuple2<String,Optional<Boolean>>> userJoinRDD = userRDD.leftOuterJoin(blockRDD);

              JavaPairRDD<String,Tuple2<String,Optional<Boolean>>> userFilterRDD =  userJoinRDD.filter(new Function<Tuple2<String, Tuple2<String, Optional<Boolean>>>, Boolean>() {
                    @Override
                    public Boolean call(Tuple2<String, Tuple2<String, Optional<Boolean>>> v1) throws Exception {
                        if(v1._2._2.isPresent()&&v1._2._2.get()){

                            return false;
                        }else{
                            return true;
                        }

                    }
                });

              JavaRDD<String> userLogRDD =  userFilterRDD.map(new Function<Tuple2<String,Tuple2<String,Optional<Boolean>>>, String>() {
                    @Override
                    public String call(Tuple2<String, Tuple2<String, Optional<Boolean>>> v1) throws Exception {
                        return v1._2._1;
                    }
                });
                return userLogRDD;
            }
        });

        userLogDStream.print();

        ssc.start();
        ssc.awaitTermination();
        ssc.close();
    }
}
